package flink.cdc;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkCDC
{
    public static void main(String[] args) throws Exception
    {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        // 开启 Checkpoint检查点 并指定状态后端为 HDFS     memory fs rocksdb
//        // 5 秒检查一次
//        env.enableCheckpointing(5000L);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        // 失败重试策略，新版本可以使用默认值
////        env.setRestartStrategy(RestartStrategies.failureRateRestart());
//        // 指定状态后端
//        env.setStateBackend(new FsStateBackend("hdfs://127.0.0.1:8020/gmall-flink/ck"));

        // 创建 Flink-MySQL-CDC 的 Source
        DebeziumSourceFunction<String> mysqlSource = MySQLSource.<String>builder()
                .hostname("localhost")
                .port(3306)
                .username("root")
                .password("root")
                .databaseList("gmall-flink")
                //可选配置项,如果不指定该参数,则会读取上一个配置下的所有表的数据，注意：指定的时候需要使用"db.table"的方式
                .tableList("gmall-flink.base_category1")
                // initial: 初始化历史数据并监听binlog
                // latest:监听binlog
                .startupOptions(StartupOptions.initial())
                // 反序列化器
                .deserializer(new StringDebeziumDeserializationSchema())
                .build();

        // 使用 CDC Source 从 MySQL 读取数据
        DataStreamSource<String> resultStream = env.addSource(mysqlSource);

        // 打印数据
        resultStream.print();

        // 执行任务
        env.execute("FlinkCDC");
    }
}
